๐๏ธGitะฏัะฐ๐๏ธ
Commit 80f91b3457ec5d217952fce18b7f043b3e4da5ee
Parents : fd8414a
Author : simulationstation <32910678+simulationstation@users.noreply.github.com>
Signature : Signature validation error
Date : 2026-08-11T16:00:37-10:00
Committer : GitHub <noreply@github.com>
Date : 2026-08-12T02:00:37Z
fix(mqtt): isolate overlapping client sessions (#6616)
Changes
2 files changed, 266 insertions(+), 90 deletions(-)
Diff
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt
index 8085c2c03f..64558313f3 100644
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt
+++ b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt
@@ -17,17 +17,24 @@
package org.meshtastic.core.network.repository
import co.touchlab.kermit.Logger
+import kotlinx.atomicfu.atomic
import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.channels.ProducerScope
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.StateFlow
-import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.flatMapLatest
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
@@ -60,7 +67,6 @@ import org.meshtastic.mqtt.transport.ws.WebSocketTransportFactory
import org.meshtastic.proto.ModuleConfig
import org.meshtastic.proto.MqttClientProxyMessage
import org.meshtastic.proto.ServiceEnvelope
-import kotlin.concurrent.Volatile
import kotlin.uuid.Uuid
@Single(binds = [MQTTRepository::class])
@@ -96,26 +102,26 @@ class MQTTRepositoryImpl(
private const val RECONNECT_BACKOFF_MULTIPLIER = 2
}
- @Volatile private var client: MqttClientSession? = null
private var mqttClientFactory: (MqttClientSetup) -> MqttClientSession = ::defaultMqttClientFactory
+ private val scope = CoroutineScope(dispatchers.default + SupervisorJob())
+ private val activeSession = MutableStateFlow<ActiveMqttSession?>(null)
- private val _connectionState = MutableStateFlow<ConnectionState>(ConnectionState.Disconnected.Idle)
- override val connectionState: StateFlow<ConnectionState> = _connectionState.asStateFlow()
+ override val connectionState: StateFlow<ConnectionState> =
+ activeSession
+ .flatMapLatest { session -> session?.client?.connectionState ?: flowOf(ConnectionState.Disconnected.Idle) }
+ .onEach(::logConnectionState)
+ .stateIn(scope, SharingStarted.Eagerly, ConnectionState.Disconnected.Idle)
@OptIn(ExperimentalSerializationApi::class)
private val json = Json {
ignoreUnknownKeys = true
exceptionsWithDebugInfo = false
}
- private val scope = CoroutineScope(dispatchers.default + SupervisorJob())
private val publishSemaphore = Semaphore(20)
override fun disconnect() {
Logger.i { "MQTT Disconnecting" }
- val c = client
- client = null
- _connectionState.value = ConnectionState.Disconnected.Idle
- scope.launch { safeCatching { c?.close() }.onFailure { e -> Logger.w(e) { "MQTT clean disconnect failed" } } }
+ closeSession(takeActiveSession())
}
// json_enabled is deprecated in the protobuf schema but remains the only way to toggle MQTT JSON
@@ -145,7 +151,8 @@ class MQTTRepositoryImpl(
logLevel = if (buildConfigProvider.isDebug) MqttLogLevel.DEBUG else MqttLogLevel.WARN,
),
)
- client = newClient
+ val session = ActiveMqttSession(newClient)
+ closeSession(replaceActiveSession(session))
val subscriptions: List<Subscription> = buildList {
channelSet.subscribeList.forEach { globalId ->
@@ -179,63 +186,97 @@ class MQTTRepositoryImpl(
// that arrive immediately after SUBSCRIBE.
launch { newClient.messages.collect { msg -> processMessage(msg) } }
- // Forward the client's connection state to the repo-level StateFlow for UI observation.
- // Also emit structured log messages on transitions so reconnect attempt counts and
- // disconnect reason codes are visible in Crashlytics/Datadog without any PII.
- launch { newClient.connectionState.collect { state -> updateConnectionState(state) } }
-
// Retry the initial connect with exponential backoff. Once established,
// autoReconnect handles subsequent drops and re-subscribes internally.
- launch {
- var reconnectDelay = INITIAL_RECONNECT_DELAY_MS
- while (true) {
- val result = safeCatching {
- Logger.i {
- if (buildConfigProvider.isDebug) "MQTT Connecting to $endpoint" else "MQTT Connecting..."
- }
- newClient.connect(endpoint)
- if (subscriptions.isNotEmpty()) {
- Logger.d { "MQTT subscribing to ${subscriptions.size} topics" }
- newClient.subscribe(subscriptions)
- }
- Logger.i { "MQTT connected and subscribed" }
- }
- val failure = result.exceptionOrNull()
- when {
- result.isSuccess -> return@launch
-
- failure is MqttException.ConnectionRejected && failure.isCredentialRejection() -> {
- Logger.e(failure) { "MQTT connection rejected (unrecoverable), stopping" }
- close(failure)
- return@launch
+ val connectJob =
+ launch(start = CoroutineStart.LAZY) {
+ var reconnectDelay = INITIAL_RECONNECT_DELAY_MS
+ while (isActiveSession(session)) {
+ val result = safeCatching {
+ if (!isActiveSession(session)) return@launch
+ Logger.i {
+ if (buildConfigProvider.isDebug) "MQTT Connecting to $endpoint" else "MQTT Connecting..."
+ }
+ newClient.connect(endpoint)
+ if (!isActiveSession(session)) return@launch
+ if (subscriptions.isNotEmpty()) {
+ Logger.d { "MQTT subscribing to ${subscriptions.size} topics" }
+ newClient.subscribe(subscriptions)
+ }
+ Logger.i { "MQTT connected and subscribed" }
}
+ val failure = result.exceptionOrNull()
+ when {
+ result.isSuccess -> return@launch
+
+ failure is MqttException.ConnectionRejected && failure.isCredentialRejection() -> {
+ Logger.e(failure) { "MQTT connection rejected (unrecoverable), stopping" }
+ close(failure)
+ return@launch
+ }
- else -> {
- // Broker- and network-side failures are what this retry loop exists to absorb โ an
- // unreachable host, a TLS problem, a dropped connection, or a broker that violates the
- // MQTT 5 spec (e.g. the topic-alias limit). None are defects in this app, and reporting
- // every retry as a non-fatal drowned real regressions.
- //
- // Anything else landing here is unexpected โ a fault in our own connect/subscribe setup
- // rather than the peer's โ so it keeps reporting.
- if (failure.isExpectedMqttRetryFailure()) {
- Logger.w(failure) { "MQTT connect failed, retrying in ${reconnectDelay}ms" }
- } else {
- Logger.e(failure) { "MQTT connect failed unexpectedly, retrying in ${reconnectDelay}ms" }
+ else -> {
+ if (!isActiveSession(session)) return@launch
+ // Broker- and network-side failures are what this retry loop exists to absorb โ an
+ // unreachable host, a TLS problem, a dropped connection, or a broker that violates the
+ // MQTT 5 spec (e.g. the topic-alias limit). None are defects in this app, and reporting
+ // every retry as a non-fatal drowned real regressions.
+ //
+ // Anything else landing here is unexpected โ a fault in our own connect/subscribe setup
+ // rather than the peer's โ so it keeps reporting.
+ if (failure.isExpectedMqttRetryFailure()) {
+ Logger.w(failure) { "MQTT connect failed, retrying in ${reconnectDelay}ms" }
+ } else {
+ Logger.e(failure) {
+ "MQTT connect failed unexpectedly, retrying in ${reconnectDelay}ms"
+ }
+ }
+ delay(reconnectDelay)
+ reconnectDelay =
+ (reconnectDelay * RECONNECT_BACKOFF_MULTIPLIER).coerceAtMost(MAX_RECONNECT_DELAY_MS)
}
- delay(reconnectDelay)
- reconnectDelay =
- (reconnectDelay * RECONNECT_BACKOFF_MULTIPLIER).coerceAtMost(MAX_RECONNECT_DELAY_MS)
}
}
}
+ session.connectJob.value = connectJob
+ if (isActiveSession(session)) {
+ connectJob.start()
+ } else {
+ session.connectJob.getAndSet(null)?.cancel()
+ }
+
+ awaitClose {
+ activeSession.compareAndSet(session, null)
+ closeSession(session)
+ }
+ }
+
+ private fun replaceActiveSession(replacement: ActiveMqttSession): ActiveMqttSession? {
+ while (true) {
+ val current = activeSession.value
+ if (activeSession.compareAndSet(current, replacement)) return current
}
+ }
- awaitClose { disconnect() }
+ private fun takeActiveSession(): ActiveMqttSession? {
+ while (true) {
+ val current = activeSession.value ?: return null
+ if (activeSession.compareAndSet(current, null)) return current
+ }
}
- internal fun updateConnectionState(state: ConnectionState) {
- _connectionState.value = state
+ private fun closeSession(session: ActiveMqttSession?) {
+ if (session == null || !session.closeStarted.compareAndSet(expect = false, update = true)) return
+ session.connectJob.getAndSet(null)?.cancel()
+ scope.launch {
+ safeCatching { session.client.close() }.onFailure { e -> Logger.w(e) { "MQTT clean disconnect failed" } }
+ }
+ }
+
+ private fun isActiveSession(session: ActiveMqttSession): Boolean =
+ !session.closeStarted.value && activeSession.value === session
+
+ private fun logConnectionState(state: ConnectionState) {
when (state) {
ConnectionState.Connecting -> Logger.i { "MQTT connecting" }
@@ -289,21 +330,21 @@ class MQTTRepositoryImpl(
}
override fun publish(topic: String, data: ByteArray, retained: Boolean) {
- val currentClient = client
- if (currentClient == null) {
- Logger.w {
- if (buildConfigProvider.isDebug) {
- "MQTT publish to $topic dropped: client not connected"
- } else {
- "MQTT publish dropped: client not connected"
- }
- }
- return
- }
scope.launch {
publishSemaphore.withPermit {
+ val session = activeSession.value
+ if (session == null || !isActiveSession(session)) {
+ Logger.w {
+ if (buildConfigProvider.isDebug) {
+ "MQTT publish to $topic dropped: client not connected"
+ } else {
+ "MQTT publish dropped: client not connected"
+ }
+ }
+ return@withPermit
+ }
safeCatching {
- currentClient.publish(
+ session.client.publish(
MqttMessage(topic = topic, payload = data, qos = QoS.AT_LEAST_ONCE, retain = retained),
)
}
@@ -321,6 +362,11 @@ class MQTTRepositoryImpl(
}
}
+private class ActiveMqttSession(val client: MqttClientSession) {
+ val closeStarted = atomic(false)
+ val connectJob = atomic<Job?>(null)
+}
+
/**
* `true` only for CONNACK reason codes where retrying can never help โ the broker examined our credentials or client
* identity and refused them.
diff --git a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt
index 50caf893f9..dbb2612848 100644
--- a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt
+++ b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt
@@ -20,7 +20,7 @@ import dev.mokkery.MockMode
import dev.mokkery.answering.returns
import dev.mokkery.every
import dev.mokkery.mock
-import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.Job
import kotlinx.coroutines.async
@@ -460,38 +460,157 @@ class MQTTRepositoryImplTest {
}
@Test
- fun `connection state flow reflects repository state updates`() {
+ fun `connection state flow reflects active client state updates`() = runTest {
+ val harness = createHarness()
+ val collector = startProxyCollection(harness.repository)
+ runCurrent()
+ val disconnectError = MqttException.ConnectionLost(ReasonCode.UNSPECIFIED_ERROR, "link lost")
+
+ assertEquals(ConnectionState.Disconnected.Idle, harness.repository.connectionState.value)
+
+ harness.client.emitState(ConnectionState.Connecting)
+ runCurrent()
+ assertEquals(ConnectionState.Connecting, harness.repository.connectionState.value)
+
+ harness.client.emitState(ConnectionState.Connected)
+ runCurrent()
+ assertEquals(ConnectionState.Connected, harness.repository.connectionState.value)
+
+ harness.client.emitState(ConnectionState.Reconnecting(attempt = 2, lastError = disconnectError))
+ runCurrent()
+ val reconnecting = assertIs<ConnectionState.Reconnecting>(harness.repository.connectionState.value)
+ assertEquals(2, reconnecting.attempt)
+ assertEquals("link lost", reconnecting.lastError?.message)
+
+ harness.client.emitState(ConnectionState.Disconnected(reason = disconnectError))
+ runCurrent()
+ val disconnected = assertIs<ConnectionState.Disconnected>(harness.repository.connectionState.value)
+ assertEquals("link lost", disconnected.reason?.message)
+
+ collector.cancelAndJoin()
+ runCurrent()
+ }
+
+ @Test
+ fun `stale collector teardown closes only its session and preserves replacement state`() = runTest {
+ val firstClient = FakeMqttClientSession()
+ val replacementClient = FakeMqttClientSession()
+ val clients = ArrayDeque(listOf(firstClient, replacementClient))
+ val dispatcher = kotlinx.coroutines.test.StandardTestDispatcher(testScheduler)
val repository =
MQTTRepositoryImpl(
- radioConfigRepository = FakeRadioConfigRepository(),
+ radioConfigRepository = defaultRadioConfigRepository(),
nodeRepository = FakeNodeRepository().apply { setMyId("!12345678") },
buildConfigProvider = buildConfigProvider,
- dispatchers =
- CoroutineDispatchers(
- io = Dispatchers.Default,
- main = Dispatchers.Default,
- default = Dispatchers.Default,
- ),
- mqttClientFactory = { FakeMqttClientSession() },
+ dispatchers = CoroutineDispatchers(io = dispatcher, main = dispatcher, default = dispatcher),
+ mqttClientFactory = { clients.removeFirst() },
)
- val disconnectError = MqttException.ConnectionLost(ReasonCode.UNSPECIFIED_ERROR, "link lost")
- assertEquals(ConnectionState.Disconnected.Idle, repository.connectionState.value)
+ val firstCollector = startProxyCollection(repository)
+ runCurrent()
+ firstClient.emitState(ConnectionState.Connected)
+ runCurrent()
+ assertEquals(ConnectionState.Connected, repository.connectionState.value)
- repository.updateConnectionState(ConnectionState.Connecting)
+ val replacementCollector = startProxyCollection(repository)
+ runCurrent()
+ replacementClient.emitState(ConnectionState.Connecting)
+ runCurrent()
+
+ assertEquals(1, firstClient.closeCalls, "installing a replacement must close the displaced session")
+ assertEquals(0, replacementClient.closeCalls)
assertEquals(ConnectionState.Connecting, repository.connectionState.value)
- repository.updateConnectionState(ConnectionState.Connected)
+ firstClient.emitState(ConnectionState.Connected)
+ firstCollector.cancelAndJoin()
+ runCurrent()
+
+ assertEquals(1, firstClient.closeCalls, "stale awaitClose must not close its client twice")
+ assertEquals(0, replacementClient.closeCalls, "stale awaitClose must not close the replacement")
+ assertEquals(ConnectionState.Connecting, repository.connectionState.value, "stale state must not win")
+
+ replacementClient.emitState(ConnectionState.Connected)
+ runCurrent()
assertEquals(ConnectionState.Connected, repository.connectionState.value)
- repository.updateConnectionState(ConnectionState.Reconnecting(attempt = 2, lastError = disconnectError))
- val reconnecting = assertIs<ConnectionState.Reconnecting>(repository.connectionState.value)
- assertEquals(2, reconnecting.attempt)
- assertEquals("link lost", reconnecting.lastError?.message)
+ replacementCollector.cancelAndJoin()
+ runCurrent()
+ assertEquals(1, replacementClient.closeCalls)
+ assertEquals(ConnectionState.Disconnected.Idle, repository.connectionState.value)
+ }
- repository.updateConnectionState(ConnectionState.Disconnected(reason = disconnectError))
- val disconnected = assertIs<ConnectionState.Disconnected>(repository.connectionState.value)
- assertEquals("link lost", disconnected.reason?.message)
+ @Test
+ fun `replacing a session cancels its delayed connection retry`() = runTest {
+ val firstClient = FakeMqttClientSession()
+ val replacementClient = FakeMqttClientSession()
+ firstClient.failConnectWith(MqttException.ConnectionLost(ReasonCode.UNSPECIFIED_ERROR, "offline"))
+ val clients = ArrayDeque(listOf(firstClient, replacementClient))
+ val dispatcher = kotlinx.coroutines.test.StandardTestDispatcher(testScheduler)
+ val repository =
+ MQTTRepositoryImpl(
+ radioConfigRepository = defaultRadioConfigRepository(),
+ nodeRepository = FakeNodeRepository().apply { setMyId("!12345678") },
+ buildConfigProvider = buildConfigProvider,
+ dispatchers = CoroutineDispatchers(io = dispatcher, main = dispatcher, default = dispatcher),
+ mqttClientFactory = { clients.removeFirst() },
+ )
+
+ val firstCollector = startProxyCollection(repository)
+ runCurrent()
+ assertEquals(1, firstClient.connectCalls.size)
+
+ val replacementCollector = startProxyCollection(repository)
+ runCurrent()
+ assertEquals(1, firstClient.closeCalls)
+ assertEquals(1, replacementClient.connectCalls.size)
+
+ advanceTimeBy(1_000)
+ runCurrent()
+ assertEquals(1, firstClient.connectCalls.size, "retired session must not reconnect after its retry delay")
+
+ firstCollector.cancelAndJoin()
+ replacementCollector.cancelAndJoin()
+ runCurrent()
+ }
+
+ @Test
+ fun `queued publish resolves the active session after waiting for a permit`() = runTest {
+ val firstClient = FakeMqttClientSession()
+ val replacementClient = FakeMqttClientSession()
+ val clients = ArrayDeque(listOf(firstClient, replacementClient))
+ val dispatcher = kotlinx.coroutines.test.StandardTestDispatcher(testScheduler)
+ val repository =
+ MQTTRepositoryImpl(
+ radioConfigRepository = defaultRadioConfigRepository(),
+ nodeRepository = FakeNodeRepository().apply { setMyId("!12345678") },
+ buildConfigProvider = buildConfigProvider,
+ dispatchers = CoroutineDispatchers(io = dispatcher, main = dispatcher, default = dispatcher),
+ mqttClientFactory = { clients.removeFirst() },
+ )
+ val publishGate = CompletableDeferred<Unit>()
+ firstClient.blockPublishesUntil(publishGate)
+ val firstCollector = startProxyCollection(repository)
+ runCurrent()
+
+ repeat(20) { index -> repository.publish("busy/$index", byteArrayOf(index.toByte()), retained = false) }
+ runCurrent()
+ assertEquals(20, firstClient.publishStarted.size)
+
+ repository.publish("queued/after-replacement", byteArrayOf(42), retained = false)
+ runCurrent()
+ assertEquals(20, firstClient.publishStarted.size, "the target publish must still be waiting for a permit")
+
+ val replacementCollector = startProxyCollection(repository)
+ runCurrent()
+ publishGate.complete(Unit)
+ runCurrent()
+
+ assertFalse(firstClient.publishedMessages.any { it.topic == "queued/after-replacement" })
+ assertTrue(replacementClient.publishedMessages.any { it.topic == "queued/after-replacement" })
+
+ firstCollector.cancelAndJoin()
+ replacementCollector.cancelAndJoin()
+ runCurrent()
}
// region MqttJsonPayload โ keep the existing JSON contract tests.
@@ -668,11 +787,14 @@ class MQTTRepositoryImplTest {
override val connectionState = MutableStateFlow<ConnectionState>(ConnectionState.Disconnected.Idle)
val connectCalls = mutableListOf<MqttEndpoint>()
val subscribeCalls = mutableListOf<List<Subscription>>()
+ val publishStarted = mutableListOf<MqttMessage>()
+ val publishedMessages = mutableListOf<MqttMessage>()
var closeCalls = 0
private set
private val connectFailures = ArrayDeque<Throwable>()
private val subscribeFailures = ArrayDeque<Throwable>()
+ private var publishGate: CompletableDeferred<Unit>? = null
override suspend fun connect(endpoint: MqttEndpoint) {
connectCalls += endpoint
@@ -684,7 +806,11 @@ class MQTTRepositoryImplTest {
if (subscribeFailures.isNotEmpty()) throw subscribeFailures.removeFirst()
}
- override suspend fun publish(message: MqttMessage) = Unit
+ override suspend fun publish(message: MqttMessage) {
+ publishStarted += message
+ publishGate?.await()
+ publishedMessages += message
+ }
override suspend fun close() {
closeCalls += 1
@@ -698,6 +824,10 @@ class MQTTRepositoryImplTest {
subscribeFailures.addLast(throwable)
}
+ fun blockPublishesUntil(gate: CompletableDeferred<Unit>) {
+ publishGate = gate
+ }
+
suspend fun emitMessage(message: MqttMessage) {
mutableMessages.emit(message)
}
Served by rngit 1.5.4 - Generated in 0.04s